package com.intct.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.util.HBaseUtil;
import com.intct.util.JedisUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;
import redis.clients.jedis.Jedis;

import java.util.Collections;

/**
 * @author gufg
 * @since 2025-08-22 11:35
 */
public class LookupAsyncFunc extends RichAsyncFunction<JSONObject, String> {

    private Jedis jedis;
    private AsyncConnection asyncHBaseConn;

    private String namespace;
    private String tableName;
    // 主流中ID字段
    private String id;


    public LookupAsyncFunc(String namespace, String tableName, String id) {
        this.id = id;
        this.namespace = namespace;
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = JedisUtil.getJedis();
        asyncHBaseConn = HBaseUtil.getAsyncConnection();
    }

    @Override
    public void asyncInvoke(JSONObject input, ResultFuture<String> resultFuture) throws Exception {

        // 主键
        String rowkey = input.getString(id);

        // 根据主键从Redis查询
        String getRedisGet = jedis.get(rowkey);
        String hbaseJsonStr = null;

        // 如果Redis中不存时
        if (getRedisGet == null) {
            // 查询HBase
            hbaseJsonStr = HBaseUtil.getRowkey(asyncHBaseConn, namespace, tableName, rowkey);

            // HBase存在数据
            if (JSON.parseObject(hbaseJsonStr).size() > 0) {
                // 写入Redis
                jedis.setex(rowkey, (60 * 60 * 24 * 7L), hbaseJsonStr);
                // 再次获取Redis
                getRedisGet = jedis.get(rowkey);
            }
        }

        if (getRedisGet != null) {
            // 组合(主流 + 查询维表)
            JSONObject jsonObj = JSON.parseObject(getRedisGet);
            input.putAll(jsonObj);
        }

        // 返回
        resultFuture.complete(Collections.singleton(input.toJSONString()));
    }

//    @Override
//    public void timeout(JSONObject input, ResultFuture<String> resultFuture) throws Exception {
//        System.out.println("连接HBase或Redis超时，请联系管理员！");
//    }

    @Override
    public void close() throws Exception {
        JedisUtil.close(jedis);
        HBaseUtil.closeAsycnConnection(asyncHBaseConn);
    }
}
